package com.lagou.rpc.client.netty;

import com.lagou.rpc.client.RpcClientHandler;
import com.lagou.rpc.entity.RpcResponse;
import com.lagou.rpc.protocol.JSONSerializer;
import com.lagou.rpc.protocol.RpcDecoder;
import com.lagou.rpc.protocol.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 参考了dubbo的 org.apache.dubbo.remoting.transport.netty4.NettyClient
 */
public class NettyClient {

    String url;
    String host;
    int port;
    private volatile Channel channel;
    private volatile boolean close;
    private volatile RpcClientHandler clientHandler;
    private Logger log = LoggerFactory.getLogger(NettyClient.class);

    public NettyClient(String url) {
        this.url = url;
        String[] arr = url.split(":");
        host = arr[0];
        port = Integer.valueOf(arr[1]);
        this.connect();
    }

    /**
     * 连接了netty服务器
     */
    public void connect() {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        this.clientHandler =  new RpcClientHandler();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(
                                65536, 0, 4, 0, 0));
                        pipeline.addLast(new RpcEncoder(new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(new JSONSerializer(), RpcResponse.class));
                        pipeline.addLast(clientHandler);
                    }
                });
        ChannelFuture future = bootstrap.connect(host, port);
        boolean ret = future.awaitUninterruptibly(5000, TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            this.channel = future.channel();
            this.close = false;
            log.info("连接服务器成功:" + this.url);
        }
    }

    public RpcClientHandler getClientHandler() {
        return clientHandler;
    }

    public void close() {
        this.close = true;
        if (hasChannel()) {
            this.channel.close();
        }
    }

    public boolean hasChannel() {
        return this.channel != null;
    }
    public boolean isClosed() {
        return !channel.isActive();
    }
}
